package com.zyg.stream;

import com.zyg.collector.MyCollector;
import com.zyg.pojo.User;

import java.io.IOException;
import java.nio.file.Files;
import java.nio.file.Paths;
import java.util.*;
import java.util.stream.Collectors;
import java.util.stream.Stream;

public class MyStreamDemo {
    public static void main(String[] args) throws IOException {
        //1.筛选函数 filter() distinct() 都属于中间操作 filter无状态   distinct有状态
        //获取年龄20岁以下的用户 filter
        /**
         * 源码通过Predicate实现数据的过滤
         *   @Override
         *     public final Stream<P_OUT> filter(Predicate<? super P_OUT> predicate) {
         *         Objects.requireNonNull(predicate);
         *         return new StatelessOp<P_OUT, P_OUT>(this, StreamShape.REFERENCE,
         *                                      StreamOpFlag.NOT_SIZED) {
         *             @Override
         *             Sink<P_OUT> opWrapSink(int flags, Sink<P_OUT> sink) {
         *                 return new Sink.ChainedReference<P_OUT, P_OUT>(sink) {
         *                     @Override
         *                     public void begin(long size) {
         *                         downstream.begin(-1);
         *                     }
         *
         *                     @Override
         *                     public void accept(P_OUT u) {
         *                         if (predicate.test(u))
         *                             downstream.accept(u);
         *                     }
         *                 };
         *             }
         *         };
         *     }
         */
        List<User> users = new ArrayList<>();
        users.add(new User(1,"张三",18,true,81));
        users.add(new User(2,"李四",20,true,85));
        users.add(new User(3,"王五",19,false,84));
        users.add(new User(4,"赵六",18,false,82));
        users.add(new User(5,"钱七",23,true,89));
        users.add(new User(5,"钱七",23,false,75));
        users.add(new User(6,"孙策",25,false,90));
        List<User> collect = users.stream().filter(user -> user.getAge() < 20).collect(Collectors.toList());
        System.out.println(collect);
        /**
         * 对对象去重注意实现 equals 和 hashcode 方法
         *  源码：通过reduce方法重新封装数据到一个LinkedHashSet容器中实现去重
         *    <P_IN> Node<T> reduce(PipelineHelper<T> helper, Spliterator<P_IN> spliterator) {
         *                 // If the stream is SORTED then it should also be ORDERED so the following will also
         *                 // preserve the sort order
         *                 TerminalOp<T, LinkedHashSet<T>> reduceOp
         *                         = ReduceOps.<T, LinkedHashSet<T>>makeRef(LinkedHashSet::new, LinkedHashSet::add,
         *                                                                  LinkedHashSet::addAll);
         *                 return Nodes.node(reduceOp.evaluateParallel(helper, spliterator));
         *             }
         *
         *             @Override
         *             <P_IN> Node<T> opEvaluateParallel(PipelineHelper<T> helper,
         *                                               Spliterator<P_IN> spliterator,
         *                                               IntFunction<T[]> generator) {
         *                 if (StreamOpFlag.DISTINCT.isKnown(helper.getStreamAndOpFlags())) {
         *                     // No-op
         *                     return helper.evaluate(spliterator, false, generator);
         *                 }
         *                 else if (StreamOpFlag.ORDERED.isKnown(helper.getStreamAndOpFlags())) {
         *                     return reduce(helper, spliterator);
         *                 }
         *                 else {
         *                     // Holder of null state since ConcurrentHashMap does not support null values
         *                     AtomicBoolean seenNull = new AtomicBoolean(false);
         *                     ConcurrentHashMap<T, Boolean> map = new ConcurrentHashMap<>();
         *                     TerminalOp<T, Void> forEachOp = ForEachOps.makeRef(t -> {
         *                         if (t == null)
         *                             seenNull.set(true);
         *                         else
         *                             map.putIfAbsent(t, Boolean.TRUE);
         *                     }, false);
         *                     forEachOp.evaluateParallel(helper, spliterator);
         *
         *                     // If null has been seen then copy the key set into a HashSet that supports null values
         *                     // and add null
         *                     Set<T> keys = map.keySet();
         *                     if (seenNull.get()) {
         *                         // TODO Implement a more efficient set-union view, rather than copying
         *                         keys = new HashSet<>(keys);
         *                         keys.add(null);
         *                     }
         *                     return Nodes.node(keys);
         *                 }
         *     }
         *
         *             Sink<T> opWrapSink(int flags, Sink<T> sink) {
         *                 Objects.requireNonNull(sink);
         *
         *                 if (StreamOpFlag.DISTINCT.isKnown(flags)) {
         *                     return sink;
         *                 } else if (StreamOpFlag.SORTED.isKnown(flags)) {
         *                     return new Sink.ChainedReference<T, T>(sink) {
         *                         boolean seenNull;
         *                         T lastSeen;
         *
         *                         @Override
         *                         public void begin(long size) {
         *                             seenNull = false;
         *                             lastSeen = null;
         *                             downstream.begin(-1);
         *                         }
         *
         *                         @Override
         *                         public void end() {
         *                             seenNull = false;
         *                             lastSeen = null;
         *                             downstream.end();
         *                         }
         *
         *                         @Override
         *                         public void accept(T t) {
         *                             if (t == null) {
         *                                 if (!seenNull) {
         *                                     seenNull = true;
         *                                     downstream.accept(lastSeen = null);
         *                                 }
         *                             } else if (lastSeen == null || !t.equals(lastSeen)) {
         *                                 downstream.accept(lastSeen = t);
         *                             }
         *                         }
         *                     };
         *                 } else {
         *                     return new Sink.ChainedReference<T, T>(sink) {
         *                         Set<T> seen;
         *
         *                         @Override
         *                         public void begin(long size) {
         *                             seen = new HashSet<>();
         *                             downstream.begin(-1);
         *                         }
         *
         *                         @Override
         *                         public void end() {
         *                             seen = null;
         *                             downstream.end();
         *                         }
         *
         *                         @Override
         *                         public void accept(T t) {
         *                             if (!seen.contains(t)) {
         *                                 seen.add(t);
         *                                 downstream.accept(t);
         *                             }
         *                         }
         *                     };
         *                 }
         *             }
         *
         */

        List<User> collect1 = users.stream().distinct().collect(Collectors.toList());
        System.out.println(collect1);

        //2.切片方法 limit() 和 skip() 方法 都属于有状态的中间操作
        /**
         * skip() 和 limit() 源码均是通过截取器截取指定位置和数量的字符串
         *
         * limit()
         *  public final Stream<P_OUT> limit(long maxSize) {
         *         if (maxSize < 0)
         *             throw new IllegalArgumentException(Long.toString(maxSize));
         *         return SliceOps.makeRef(this, 0, maxSize);
         *     }
         *
         * skip()
         *  public final Stream<P_OUT> skip(long n) {
         *         if (n < 0)
         *             throw new IllegalArgumentException(Long.toString(n));
         *         if (n == 0)
         *             return this;
         *         else
         *             return SliceOps.makeRef(this, n, -1);
         *     }
         *
         *   切割器方法
         *  Spliterator<T> unorderedSkipLimitSpliterator(Spliterator<T> s,
         *                                                          long skip, long limit, long sizeIfKnown) {
         *                 if (skip <= sizeIfKnown) {
         *                     // Use just the limit if the number of elements
         *                     // to skip is <= the known pipeline size
         *                     limit = limit >= 0 ? Math.min(limit, sizeIfKnown - skip) : sizeIfKnown - skip;
         *                     skip = 0;
         *                 }
         *                 return new StreamSpliterators.UnorderedSliceSpliterator.OfRef<>(s, skip, limit);
         *             }
         *
         */
        //limit截取前几位,从头开始截取
        List<User> collect2 = users.stream().limit(3).collect(Collectors.toList());
        System.out.println(collect2);
        //skip表示跳过几位，然后开始截取
        List<User> collect3 = users.stream().skip(2).collect(Collectors.toList());
        System.out.println(collect3);
        List<User> collect4 = users.stream().skip(1).limit(3).collect(Collectors.toList());
        List<User> collect5 = users.stream().limit(3).skip(1).collect(Collectors.toList());
        //注意collect4和collect5的区别
        System.out.println(collect4);
        System.out.println(collect5);
        // 3.映射操作map() 无状态中间操作
        /**
         *  public final <R> Stream<R> map(Function<? super P_OUT, ? extends R> mapper) {
         *         Objects.requireNonNull(mapper);
         *         return new StatelessOp<P_OUT, R>(this, StreamShape.REFERENCE,
         *                                      StreamOpFlag.NOT_SORTED | StreamOpFlag.NOT_DISTINCT) {
         *             @Override
         *             Sink<P_OUT> opWrapSink(int flags, Sink<R> sink) {
         *                 return new Sink.ChainedReference<P_OUT, R>(sink) {
         *                     @Override
         *                     public void accept(P_OUT u) {
         *                         downstream.accept(mapper.apply(u));
         *                     }
         *                 };
         *             }
         *         };
         *     }
         */
        List<String> collect6 = users.stream().map(User::getName).collect(Collectors.toList());
        System.out.println(collect6);
        int sum1 = Stream.of("1", "2", "6", "8", "3").mapToInt(s -> Integer.parseInt(s)).sum();
        //todo
        //Long sum2 = Stream.of("1", "2", "6", "8", "3").collect(Collectors.summingInt(Integer::sum));
        System.out.println(sum1);
        //System.out.println(sum2);
        //4.匹配操作 anyMatch() 和 allMatch()
        /**
         *
         * 原理：根据匹配的枚举类型执行不同逻辑
         *
         *
         * Do all elements match the predicate?
         * ANY(true, true),
         * Do any elements match the predicate?
         * ALL(false, false),
         *
         * public final boolean anyMatch(Predicate<? super P_OUT> predicate) {
         *    return evaluate(MatchOps.makeRef(predicate, MatchOps.MatchKind.ANY));
         *  }
         *
         *  public final boolean allMatch(Predicate<? super P_OUT> predicate) {
         *         return evaluate(MatchOps.makeRef(predicate, MatchOps.MatchKind.ALL));
         *     }
         *
         * anyMatch()
         * public static <T> TerminalOp<T, Boolean> makeRef(Predicate<? super T> predicate,
         *             MatchKind matchKind) {
         *         Objects.requireNonNull(predicate);
         *         Objects.requireNonNull(matchKind);
         *         class MatchSink extends BooleanTerminalSink<T> {
         *             MatchSink() {
         *                 super(matchKind);
         *             }
         *
         *             @Override (一次匹配成功即返回)
         *             public void accept(T t) {
         *                 if (!stop && predicate.test(t) == matchKind.stopOnPredicateMatches) {
         *                     stop = true;
         *                     value = matchKind.shortCircuitResult;
         *                 }
         *             }
         *         }
         *
         *
         *     public final boolean anyMatch(Predicate<? super P_OUT> predicate) {
         *         return evaluate(MatchOps.makeRef(predicate, MatchOps.MatchKind.ANY));
         *     }
         *
         */

        if (users.stream().anyMatch(user -> user.getAge() > 19)){
            System.out.println("短路输出  满足一个即返回true 输出");
        }
        if (users.stream().allMatch(user -> user.getAge() > 19)){
            System.out.println("短路输出  满足所有才返回true 输出");
        }

        //5.查找操作 findAny() 和 findFirst()
        /**  //判断是否是并行流，非并行流只会匹配第一个
         *
         *     final <R> R evaluate(TerminalOp<E_OUT, R> terminalOp) {
         *         assert getOutputShape() == terminalOp.inputShape();
         *         if (linkedOrConsumed)
         *             throw new IllegalStateException(MSG_STREAM_LINKED);
         *         linkedOrConsumed = true;
         *
         *         return isParallel()
         *                ? terminalOp.evaluateParallel(this, sourceSpliterator(terminalOp.getOpFlags()))
         *                : terminalOp.evaluateSequential(this, sourceSpliterator(terminalOp.getOpFlags()));
         *     }
         */
        //串行流
        Optional<User> optional1 = users.stream().filter(user -> user.getAge() < 20).findAny();
        //并行流
        Optional<User> optional2 = users.parallelStream().filter(user -> user.getAge() < 20).findAny();
        if (optional1.isPresent()){
            User user = optional1.get();
            System.out.println(user);
        }

        if (optional2.isPresent()){
            User user = optional2.get();
            System.out.println(user);
        }
        //6.归纳操作 如reduce()求和
        /**
         * 源码：combine其他流，第一次进来时先执行begin方法，然后执行accept,最后获取get
         *  public static <T> TerminalOp<T, Optional<T>> makeRef(BinaryOperator<T> operator) {
         *         Objects.requireNonNull(operator);
         *         class ReducingSink implements AccumulatingSink<T, Optional<T>, ReducingSink> {
         *             private boolean empty;
         *             private T state;
         *
         *            //修改empty状态为true,状态为null(初始化)
         *             public void begin(long size) {
         *                 empty = true;
         *                 state = null;
         *             }
         *
         *             //如果empty为true,修改为false，将当前值传递给state
         *             //empty不为空，将上次的state与当前传入的值进行运算
                      public void accept(T t) {
         *                 if (empty) {
         *                     empty = false;
         *                     state = t;
         *                 } else {
         *                     state = operator.apply(state, t);
         *                 }
         *             }
         *
         *             //返回当前结果值，empty为true,说明当前未返回和值
         *             public Optional<T> get() {
         *                 return empty ? Optional.empty() : Optional.of(state);
         *             }
         *
         *             public void combine(ReducingSink other) {
         *                 if (!other.empty)
         *                     accept(other.state);
         *             }
         *         }
         */
        Optional<Integer> optional3 = users.stream().map(User::getAge).distinct().reduce(Integer::sum);
        if (optional3.isPresent()){
            System.out.println(optional3.get());
        }

        //获取最大和最小值 max() 和 min()
        Optional<User> optional4 = users.stream().max(Comparator.comparing(User::getAge));
        if (optional4.isPresent()){
            System.out.println(optional4.get());
        }

        Optional<Integer> optional5 = users.stream().map(User::getAge).min(Integer::compareTo);
        if (optional5.isPresent()){
            System.out.println(optional5.get());
        }

        //基于文件构建流
        Files.list(Paths.get("D:\\com\\zyg\\test")).forEach(s-> System.out.println(s));

        Files.list(Paths.get("D:\\com\\zyg\\test")).forEach(path -> {
            try {
                Files.lines(path).forEach(content-> System.out.println(content));
            } catch (IOException e) {
                e.printStackTrace();
            }
        });
        //7. 收集器使用
        /**
         * 7.1 简单分组
         */
        Map<Integer, List<User>> collect7 = users.stream().collect(Collectors.groupingBy(User::getAge));
        System.out.println(collect7);
        /**
         * 7.2 复杂分组
         */
        //先按照年龄分组再按照男女分组
        Map<Integer, Map<Boolean, List<User>>> collect11 = users.stream().collect(Collectors.groupingBy(User::getAge, Collectors.groupingBy(User::getFemale)));
        System.out.println(collect11);

        Map<Integer, Map<String, List<User>>> collect12 = users.stream().collect(Collectors.groupingBy(User::getAge, Collectors.groupingBy(user -> {
            if (user.getFemale()) {
                return "female";
            } else {
                return "male";
            }
        })));
        System.out.println(collect12);
        Map<Integer, Map<Boolean, User>> collect13 = users.stream().collect(Collectors.groupingBy(User::getAge, Collectors.groupingBy(User::getFemale,
                Collectors.collectingAndThen(Collectors.maxBy(Comparator.comparing(User::getScore)), Optional::get))));
        System.out.println(collect13);

        //7.2 获取最大或最小年龄的用户 以下两种方式
        Optional<User> collect8 = users.stream().collect(Collectors
               // .maxBy(Comparator.comparing(User::getAge)));
                .minBy(Comparator.comparing(User::getAge)));
        if (collect8.isPresent()){
            System.out.println(collect8.get());
        }
        Optional<User> max = users.stream()
                //.max(Comparator.comparing(User::getAge));
                .min(Comparator.comparing(User::getAge));
        if (max.isPresent()){
            System.out.println(max.get());
        }

         //7.3 获取所有用户的年龄并汇总 以下两种方式
        long sum2 = users.stream().collect(Collectors.summarizingInt(User::getAge)).getSum();
        System.out.println(sum2);
        int sum3 = users.stream().mapToInt(User::getAge).sum();
        System.out.println(sum3);
        //Double collect9 = users.stream().collect(Collectors.averagingLong(User::getAge));
        Double collect9 = users.stream().collect(Collectors.averagingInt(User::getAge));
        System.out.println(collect9);
        OptionalDouble average = users.stream().mapToInt(User::getAge).average();
        if (average.isPresent()){
            System.out.println(average.getAsDouble());
        }
        //7.4 收集器结果返回拼接
        String collect10 = users.stream().map(User::getName).collect(Collectors.joining(","));
        System.out.println(collect10);

        //自定义收集器
        List<User> collect14 = users.stream().collect(new MyCollector());
        System.out.println(collect14);

    }
}
